package com.zhentao.controller;

import lombok.SneakyThrows;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.Date;

/**
 * @Date 2024/5/28 16:26
 * @Author xdm
 **/
@RestController
public class KafkaController {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @RequestMapping("/sendTs")
    @Transactional(rollbackFor = RuntimeException.class)
    public String sendTsMsg(int i) {
        kafkaTemplate.send("topic2","消息1");

        if(i==1){
            i=1/0;
        }
        kafkaTemplate.send("topic2","消息2");

        return "事务";
    }


    @SneakyThrows
    @RequestMapping("/send")
    public String sendMsg() {

        System.out.println("开始发送");
        //send的时候能不呢知道发送失败了，应不应该知道发送失败
        //如果发送失败，消息再次处理
        //这个send方法本身就是异步处理的
       // kafkaTemplate.send("topic1", "要发送的数据" + new Date().toLocaleString());

        //可以接收到这个方法的返回值，已返回值来判断是否发送成功

        //.get（）接收到返回值 ，并且使用了get方法，这个方法就变成了同步执行；
       /* SendResult<String, String> topic1 = kafkaTemplate.send("topic1", "要发送的数据" + new Date().toLocaleString())
                .get();
        System.out.println("发送的返回值"+topic1);

       */
        //异步处理


        kafkaTemplate.send("topic1","指定key,如果key一样会被发送到同一个分区里面", "要发送的数据" + new Date().toLocaleString())
                .addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
                    @Override
                    public void onFailure(Throwable throwable) {
                        System.out.println("消息发送失败");

                        //可以把失败的信息给收集起来  --logs
                    }

                    @Override
                    public void onSuccess(SendResult<String, String> stringStringSendResult) {
                        System.out.println("消息发送成功");
                    }
                });

        System.out.println("发送结束");
        return "发送消息";
    }
}
